package com.swak.reactivex.transport.serialize;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Calendar;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.TreeSet;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import com.esotericsoftware.kryo.kryo5.Kryo;
import com.esotericsoftware.kryo.kryo5.io.Input;
import com.esotericsoftware.kryo.kryo5.io.Output;
import com.esotericsoftware.kryo.kryo5.serializers.ImmutableSerializer;
import com.swak.exception.SerializeException;
import com.swak.registry.URL;
import com.swak.spi.SPI;

/**
 * 定义 SPI 的对象
 * 
 * @author 618lf
 */
@SPI(value = "kryo")
public class KryoSerialization implements Serialization {

	private final ThreadLocal<Kryo> holder = new ThreadLocal<Kryo>() {
		@Override
		protected Kryo initialValue() {
			return create();
		}
	};

	private static Kryo create() {
		Kryo kryo = new Kryo();
		kryo.addDefaultSerializer(Throwable.class, new com.esotericsoftware.kryo.kryo5.serializers.JavaSerializer());
		kryo.register(HashMap.class);
		kryo.register(ArrayList.class);
		kryo.register(LinkedList.class);
		kryo.register(HashSet.class);
		kryo.register(TreeSet.class);
		kryo.register(Hashtable.class);
		kryo.register(Date.class);
		kryo.register(LocalDateTime.class);
		kryo.register(Calendar.class);
		kryo.register(ConcurrentHashMap.class);
		kryo.register(SimpleDateFormat.class);
		kryo.register(GregorianCalendar.class);
		kryo.register(Vector.class);
		kryo.register(BitSet.class);
		kryo.register(StringBuffer.class);
		kryo.register(StringBuilder.class);
		kryo.register(Object.class);
		kryo.register(Object[].class);
		kryo.register(String[].class);
		kryo.register(byte[].class);
		kryo.register(char[].class);
		kryo.register(int[].class);
		kryo.register(float[].class);
		kryo.register(double[].class);
		kryo.register(URL.class, new URLSerializer());
		kryo.setRegistrationRequired(false);
		kryo.setReferences(false);
		kryo.setAutoReset(true);
		kryo.setOptimizedGenerics(true);
		return kryo;
	}

	@Override
	public byte getType() {
		return Serialization.KRYO_SERIALIZATION_TYPE;
	}

	@Override
	public byte[] serialize(URL url, Object obj) throws SerializeException {
		if (obj == null) {
			return None;
		}
		Output output = null;
		try {
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			output = new Output(baos);
			holder.get().writeClassAndObject(output, obj);
			output.flush();
			return baos.toByteArray();
		} finally {
			if (output != null) {
				output.close();
			}
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public <T> T deserialize(URL url, byte[] bytes) throws SerializeException {
		if (bytes == null || bytes.length == 0) {
			return null;
		}
		Input ois = null;
		try {
			ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
			ois = new Input(bais);
			return (T) holder.get().readClassAndObject(ois);
		} finally {
			if (ois != null) {
				ois.close();
			}
		}
	}

	static class URLSerializer extends ImmutableSerializer<URL> {

		@Override
		public void write(Kryo kryo, Output output, URL object) {
			output.writeString(object.toFullString());
		}

		@Override
		public URL read(Kryo kryo, Input input, Class<? extends URL> type) {
			return URL.valueOf(input.readString());
		}
	}
}
